/*********************************************************
*
*  FJH-sim v1.0
*  Copyright � 2008 H.Y. Bruintjes, J.H. Donker
*
*  Transport layer class
*
**********************************************************/

#include "protocol/TransportLayer.h"
#include "shared/Utils.h"
#include <iostream>

/*
##################################################
Helper funcs
##################################################
*/
//these defines should use a history so it can handle overflow
#define MESSAGE_MAX				256		//1<<8
#define MESSAGENEXT(msg)		(msg = (msg+1)%MESSAGE_MAX)
#define HISTORY					100	//max HISTORY old messages, otherwise treated as messages that are newer

inline int min(int a, int b) { return a<b?a:b; }

//Why this function? If we receive an old message, we ACK it just to make sure
//the other party knows we already have this message. However, overflow occurs
//at MESSAGE_MAX so we need to check with a certain history, because we have a
//special case for messages that are _more_ recent that what we expect. There
//has to be a certain tradeoff/heuristic, which is set at HISTORY
bool before(unsigned int msg, unsigned int current) {
	if (msg < current)
		return ((current-msg) < HISTORY);
	if (msg == current)
		return false;
	//msg > current, start offsetting
	int offset = MESSAGE_MAX - (msg-current);
	return offset < HISTORY;
}

/*
##################################################
PacketWindow
##################################################
*/

PacketWindow::PacketWindow() : m_last(WINDOW_MAX), m_current(0), m_complete(false), m_hasEnd(false) {};

void PacketWindow::reset() {
	m_last = WINDOW_MAX;
	m_current = 0;
	m_complete = false;
	m_hasEnd = false;
	if (!m_contained.empty())
	{
		m_contained.clear();
	}
}

void PacketWindow::setEnd(unsigned int end) {
	m_last = end;
	m_hasEnd = true;
	if (m_current > m_last)
		m_complete = true;
}

bool PacketWindow::add(unsigned int nr) {
	if (nr < m_current || nr > m_last)
		return false;
	std::deque<unsigned int>::iterator it = std::find(m_contained.begin(), m_contained.end(), nr);
	if (it != m_contained.end())
		return false;
	m_contained.push_back(nr);
	refresh();
	return true;
}


void PacketWindow::refresh() {

	if (m_contained.size() == 0)
		return;
	std::sort(m_contained.begin(), m_contained.end());

	while(m_current == *m_contained.begin()) {
		m_current++;
		if (!m_hasEnd) {
			m_last++;
		}
		if (m_contained.size() > 0)
			m_contained.pop_front();

	}
	if (m_current > m_last)
		m_complete = true;
}

bool PacketWindow::completed() {
	return m_complete;
}

/*
##################################################
TimeoutCalc
##################################################
*/

TimeoutCalc::TimeoutCalc() : m_rtt(PACKET_TIMEOUT), m_bias(0.6f) {
}

void TimeoutCalc::addRTT(int measure) {
	m_rtt = (int)(m_bias*m_rtt + (1.0-m_bias)*measure);	//cast from float to int!
}

int TimeoutCalc::getTimeout() {
	return m_rtt + 10;	//give slack of 10
}

/*
##################################################
IncomingMessage
##################################################
*/

IncomingMessage::IncomingMessage(NodeAddress source, long long time) : m_source(source), expectedMessage(0), m_startTime(time) {
}

bool IncomingMessage::acceptData(IncomingData & data, bool isLast, long long time) {
	unsigned int nr = data.m_sequenceNumber;
	unsigned int msg = data.m_messageNumber;
	if (msg != expectedMessage) {
		if (before(msg, expectedMessage)) {
			debug_print("TransportLayer", "acceptData: message seems to have been accepted already");
			return false;//true;	//already got this some time ago
		} else {	//seems we have skipped some messages, this is bad, but just reset and take it as a new message
			debug_print("TransportLayer", "acceptData: Got message %u but expected %u, adjusting...", msg, expectedMessage);
			m_window.reset();
			parts.clear();
			expectedMessage = msg;	//everything before this message considered done
		}
	}
	bool res = m_window.add(nr);
	if (!res) {
		debug_print("TransportLayer", "acceptData: message doesnt fit in window?");
		return false;	//cannot add nr to window
	}
	if (isLast) {
		m_window.setEnd(nr);
		debug_print("TransportLayer", "Last packet for message: %u %u", nr, msg);
	}

	//add package to queue (dont sort), update window
	parts.push_back(data);	//just add at the back. Most message come in order and if they dont one big sort does the job better
	if (m_window.completed()) {
		debug_print("TransportLayer", "Complete message arrived: %u", msg);
		collectData();
		m_window.reset();
		MESSAGENEXT(expectedMessage);
		debug_print("TransportLayer", "message says: %s", messages.at(messages.size() - 1).message);
	}

	m_startTime = time;	//update timeout value, something happened
	return true;
}

void IncomingMessage::collectData() {	//call only when message is complete, causes a reset in the window
	std::sort(parts.begin(), parts.end());	//sort by sequencenumber
	int size = 0;
	for(size_t i = 0; i < parts.size(); i++) {
		size += parts.at(i).m_size;
	}
	char * message = new char[size+1];
	int offset = 0;
	//collate data
	for(size_t i = 0; i < parts.size(); i++) {
		memcpy(message+offset, parts.at(i).m_data, parts.at(i).m_size);
		offset += parts.at(i).m_size;
	}
	message[offset] = 0;
	parts.clear();
	messages.push_back(Message(size, message));
}

void IncomingMessage::update(long long time) {
	long long tick_threshold = (time - WINDOW_TIMEOUT);
	if (m_startTime < tick_threshold && parts.size() > 0) {	//only timeout when we actually received something
		//timeout, discard message
		parts.clear();
		MESSAGENEXT(expectedMessage);
		debug_print("TransportLayer", "Message receive timed out, waiting for next (%d)", expectedMessage);
	}
}

/*
##################################################
JobInfo
##################################################
*/
JobInfo::JobInfo(NodeAddress dest) : 	m_destination(dest), m_status(JOB_SUCCESS), 
				m_failureRate(0), m_nextSequenceNumber(0), m_messageNumber(0) {
}

void JobInfo::clear() {
	m_status = JOB_SUCCESS;
	m_window.reset();
	m_queue.clear();
	m_waiting.clear();
	m_failureRate = 0;
	m_nextSequenceNumber = 0;
	//dont increase messagenumber
}

void JobInfo::acceptNumber(int number) {
	m_window.add(number);
	if (m_window.completed()) {
		clear();
		MESSAGENEXT(m_messageNumber);
	}
}

void JobInfo::terminate() {
	m_status = JOB_ERROR;
	m_window.reset();
	m_queue.clear();
	m_waiting.clear();
	m_failureRate = 0;
	m_nextSequenceNumber = 0;
	MESSAGENEXT(m_messageNumber);
}


/*
##################################################
TransportLayer
##################################################
*/

TransportLayer::TransportLayer(NetworkLayer *nextLayer) : m_nextLayer(nextLayer)
{
	m_ticks = 0;
}

TransportLayer::~TransportLayer()
{
	// delete the next layer in the stack
	SAFE_DELETE(m_nextLayer);
}

bool TransportLayer::open()
{
	return m_nextLayer->open();
}

void TransportLayer::close()
{
	m_nextLayer->close();
}

// we can only have one message per destination in the buffer
int TransportLayer::send(const char* data, unsigned int length, NodeAddress destination)
{	
	JobInfo & job = m_jobs.at(getJobIndex(destination));
	// perform packet splitting
	if (job.m_status == JOB_BUSY || length == 0)
		return -1;	//busy, cant send
		
	job.clear();
	
	// put as much as possible into the queue. return the amount of data in the queue
	int dataSend = 0;
	int toSend = TRANSPORT_PAYLOAD_SIZE;
	
	length = min(length, TRANSPORT_MESSAGE_SIZE);	//cut off length if needed
	while (length > 0) {
		//debug_print("TransportLayer", "sending payload to destination %u", destination);
	
		toSend = min(length, TRANSPORT_PAYLOAD_SIZE);	//send full packet or only the last piece
		length -= toSend;
		JobPayload payload(m_ticks, job.m_nextSequenceNumber++, job.m_messageNumber, (length == 0?TTYPE_MESSAGE_LAST:TTYPE_MESSAGE));
		payload.setData(data, toSend);
		job.m_queue.push_back(payload);
		dataSend += toSend;
		data += toSend;
		
	}
	
	job.m_window.setEnd(job.m_queue.size()-1);
	
	job.m_nextSequenceNumber = 0;	//reset seqnr for next transmission, not used outside this function
	job.m_status = JOB_BUSY;
	return dataSend;
}

int TransportLayer::receive(char* data, unsigned int length, NodeAddress &source)
{
	// check for invalid packet size
	if (m_messages.size() == 0)
		return -1;	//error

	//Run through all messages from all sources untill something to receive is found or if nothing is there
	int index = -1;
	for(size_t i = 0; i < m_messages.size(); i++) {
		if (m_messages.at(i).messages.size() > 0) {
			index = (int)i;
			break;
		}
	}
	
	if (index == -1)
		return -1;	//cannot receive anything

	Message & message = m_messages.at(index).messages.at(0);	//get the message, see if we cna put it in the buffer. Anything left over will be left in the queue
	source = m_messages.at(index).m_source;

	// determine the amount of bytes we need to copy
	int toCopy = min(length, message.size);
	memcpy(data, message.message, toCopy);

	if (toCopy == message.size) {
		// buffer was large enough to contain the entire packet
		m_messages.at(index).messages.erase(m_messages.at(index).messages.begin());
	} else {
		// buffer was too small to contain the entire packet
		// copy the necessary data and reduce the packet size
		message.size -= toCopy;
		for(int i = 0; i < message.size; i++) {
			message.message[i] = message.message[i+toCopy];
		}
	}
	return toCopy;
}

int TransportLayer::select(ServiceRequest service, void * pData)
{
	// "ready to receive" request: check whether the message queue is empty
	if (service == REQUEST_READY_RECEIVE) {
		for(size_t i = 0; i < m_messages.size(); i++) {
			if (m_messages.at(i).messages.size() > 0) {
				return true;
			}
		}
		return false;
	}
	// "ready to send" request: check whether the send queue for this
	// destination is not full
	else if (service == REQUEST_READY_SEND && pData != NULL) {
		NodeAddress * pAddr = (NodeAddress*) pData;
		JobInfo & job = m_jobs.at(getJobIndex(*pAddr));
		return (job.m_status != JOB_BUSY);
	}
	// "status" request: return status of Job corresponding to
	// address.
	else if (service == REQUEST_SENDINFO) {
		NodeAddress * pAddr = (NodeAddress*) pData;
		size_t index = getJobIndex(*pAddr);
		JobInfo & job = m_jobs.at(index);
		
		int ret = job.m_status;
		
		//Do _not_ remove jobinfos. The messagenumber has to live on for the rest of the session if any new attempt is ever made to transmit to the destination
		//The destination will keep track of the last message received
		// check whether a job has completed
		// if so, remove to spare memory
		//if (job.m_status == JOB_SUCCESS || job.m_status == JOB_ERROR)
		//	m_jobs.erase(m_jobs.begin() + index);
		
		return ret;
	}
	
	return false;
}

void TransportLayer::update()
{
	// increment the tickcount
	m_ticks++;
	
	// update networklayer to perform 'stuff' (buffering etc)
	m_nextLayer->update();

	// receive any incoming data
	while (m_nextLayer->select(REQUEST_READY_RECEIVE, NULL)) {
		// while receiving..
			// put the data into the appropiate packet. If its a new packet, add it as such into the m_packets vector
			// if its an ACK: ACK a job element
			// if the job is done, update its status
		// making sure the entire payload can be received

		TransportPacket data;
		NodeAddress source;
		int res = m_nextLayer->receive((char*)&data, sizeof(TransportPacket), source);
		if (res == -1)
			continue;
		data.m_length = res - sizeof(TransportLayerHeader);	// received data - TL header = message payload size
		
		if (data.m_header.type == TTYPE_ACK) {	//do ACK stuffz
			debug_print("TransportLayer", "Received ACK");

			// find JobInfo
			JobInfo & job = m_jobs.at(getJobIndex(source));
			unsigned int msg = data.m_header.messageNumber;
			if (job.m_messageNumber != msg) {
				debug_print("TransportLayer", "    ACK was for message %d, but expecting %d, with source %u", msg, job.m_messageNumber, source);
				continue;
			} else {
				debug_print("TransportLayer", "    ACK was for message %d, seq %d, with source %u", msg, data.m_header.sequenceNumber, source);
			}
			
			int rtt = m_ticks;
			bool acceptedACK = false;
			for(size_t i = 0; i < job.m_waiting.size(); i++) {
				int number = data.m_header.sequenceNumber;
				if (job.m_waiting[i].m_packet.m_header.sequenceNumber == data.m_header.sequenceNumber) {
					rtt -= job.m_waiting[i].m_startTime;
					job.m_waiting.erase(job.m_waiting.begin() + i);
					job.acceptNumber(number);
					acceptedACK = true;
					break;
				}
			}
			
			if (!acceptedACK) {
				//Can also ACK something in the queue if it just missed the timeout
				for(size_t i = 0; i < job.m_queue.size(); i++) {
					int number = data.m_header.sequenceNumber;
					if (job.m_queue[i].m_packet.m_header.sequenceNumber == data.m_header.sequenceNumber) {
						rtt -= job.m_waiting[i].m_startTime;
						job.m_queue.erase(job.m_queue.begin() + i);
						job.acceptNumber(number);
						acceptedACK = true;
						break;
					}
				}
			}
			
			if (acceptedACK) {
				job.m_timer.addRTT(rtt);
			}
		} else if (data.m_header.type == TTYPE_MESSAGE || data.m_header.type == TTYPE_MESSAGE_LAST) {
			debug_print("TransportLayer", "Received message");

			//find IncomingMessage
			IncomingMessage & msg = m_messages.at(getIncomingIndex(source));
			
			//Generate IncomingData structure
			IncomingData dataIn;
			dataIn.m_size = data.m_length;
			memcpy(dataIn.m_data, data.m_data, dataIn.m_size);
			dataIn.m_sequenceNumber = data.m_header.sequenceNumber;
			dataIn.m_messageNumber = data.m_header.messageNumber;

			//Handle the packet
			if (msg.acceptData(dataIn, (data.m_header.type == TTYPE_MESSAGE_LAST), m_ticks)) {
				sendAck(source, data.m_header.sequenceNumber, data.m_header.messageNumber);	//send ACK if we accept it
				//sendAck(source, data.m_header.sequenceNumber, data.m_header.messageNumber);	//send ACK if we accept it
			}
			else
				debug_print("TransportLayer", "Message could not be accepted %d %d", data.m_header.sequenceNumber, data.m_header.messageNumber);
		} else {
			debug_print("TransportLayer", "received unknown");
		}
	}

	//sleepThread(100);	//little time for sending parties to change to listen mode
	
	// for each destination, send packets, ACK them and check for timeouts
	for(size_t index = 0; index < m_jobs.size(); index++) {
		JobInfo & job = m_jobs[index];
		//debug_print("TransportLayer", "Destination %d: Queued %d, Need ACK %d", job.m_destination, job.m_queue.size(), job.m_waiting.size());
				
		// send all outgoing data if possible
		// check if not too much is waiting
		//First ACKS if needed
		long long ack_threshold = (m_ticks - ACKDELAY);
		while (job.m_acking.size() > 0 && m_nextLayer->select(REQUEST_READY_SEND, NULL)) {
			JobPayload & load = job.m_acking[0];
			if (load.m_startTime < ack_threshold) {
				int res = m_nextLayer->send((char*)&(load.m_packet), load.m_packet.sendSize(), job.m_destination);
				if (res == -1) {
					debug_print("TransportLayer", "Failure sending ACK");
				} else {
					debug_print("TransportLayer", "Send ACK to %d", job.m_destination);
				}
				job.m_acking.pop_front();
			} else {
				break;	//cannot send ack (they are in orderi n the queue)
			}
			
		}
		//Then messages
		while (job.m_queue.size() > 0 && m_nextLayer->select(REQUEST_READY_SEND, NULL)) {
			JobPayload & jobload = job.m_queue.at(0);
			int res = m_nextLayer->send((char*)&(jobload.m_packet), jobload.m_packet.sendSize(), job.m_destination);
			if (res == -1) {
				debug_print("TransportLayer", "Failure sending JobPayload");
			} else {
				debug_print("TransportLayer", "Send JobPayload to %d (size %d)", job.m_destination, jobload.m_packet.sendSize());
			}
			if (jobload.m_packet.m_header.type == TTYPE_MESSAGE || jobload.m_packet.m_header.type == TTYPE_MESSAGE_LAST) {	// messages need ACK'ing
				jobload.m_startTime = m_ticks;	//reset timestamp
				job.m_waiting.push_back(jobload);
			}
			job.m_queue.erase(job.m_queue.begin());
		}

		// if something has timed out: set error status
		//debug_print("transportLayer", "TimeoutCalc suggests timeout of: %d", job.m_timer.getTimeout());
		long long tick_threshold = (m_ticks - job.m_timer.getTimeout());
		for(size_t i = 0; i < job.m_waiting.size(); ) {
			if (job.m_waiting[i].m_startTime < tick_threshold) {	//timeout
				debug_print("TransportLayer", "!!packet timeout!!");
				job.m_failureRate++;
				job.m_waiting[i].m_startTime = m_ticks;	//reset timestamp
				job.m_queue.push_back(job.m_waiting[i]);	//add packet to front of queue to maintain window properties
				job.m_waiting.erase(job.m_waiting.begin() + i);
			} else {
				i++;
			}
		}
		
		if (job.m_failureRate >= MAX_FAULTRATE && job.m_status != JOB_ERROR) {	//the destination is unreachable
			debug_print("TransportLayer", "fault rate reached");
			job.terminate();
		}
	}

	for(size_t i = 0; i < m_messages.size(); i++) {
		m_messages[i].update(m_ticks);
	}
}

size_t TransportLayer::getJobIndex(NodeAddress address) {
	for(size_t index = 0; index < m_jobs.size(); index++) {
		JobInfo & job = m_jobs.at(index);
		if (job.m_destination == address) {
			return index;
		}
	}
	
	m_jobs.push_back(JobInfo(address));
	return m_jobs.size() - 1;
}

size_t TransportLayer::getIncomingIndex(NodeAddress address) {
	for(size_t index = 0; index < m_messages.size(); index++) {
		IncomingMessage & msg = m_messages.at(index);
		if (msg.m_source == address) {
			return index;
		}
	}
	
	m_messages.push_back(IncomingMessage(address, m_ticks));
	return m_messages.size() - 1;
}

bool TransportLayer::sendAck(NodeAddress destination, int sequenceNumber, int messageNumber) {
	// don't check queuesize: ACK has priority
	JobInfo & job = m_jobs.at(getJobIndex(destination));
	JobPayload payload(m_ticks, sequenceNumber, messageNumber, TTYPE_ACK);
	//job.m_queue.push_front(payload);
	job.m_acking.push_back(payload);
	debug_print("TransportLayer", "Sending ACk for seq %d, msg %d to %d", sequenceNumber, messageNumber, destination);
	
	return true;
}
